package TableAndSQL

import org.apache.flink.streaming.api.scala._
import org.apache.flink.table.api.{DataTypes, EnvironmentSettings}
import org.apache.flink.table.api.scala._
import org.apache.flink.table.descriptors.{Csv, FileSystem, OldCsv, Schema}

/**
 * 将数据发送到MySQL
 */
object MySQLOutput {
  def main(args: Array[String]): Unit = {

    val env = StreamExecutionEnvironment.getExecutionEnvironment
    env.setParallelism(1)

    //创建一个基于blink的流式
    val BlinkStreamSetting = EnvironmentSettings.newInstance()
      .useBlinkPlanner()
      .inStreamingMode()
      .build()
    val blinkStreamTableEnv = StreamTableEnvironment.create(env, BlinkStreamSetting)

    //读取外部文件
    val filePath = "src/main/resources/SensorReading"
    //读取数据
    blinkStreamTableEnv.connect(new FileSystem().path(filePath))
      .withFormat(new Csv())
      .withSchema(new Schema()
        .field("id", DataTypes.STRING())
        .field("timestamp", DataTypes.BIGINT())
        .field("temp", DataTypes.DOUBLE())
      ).createTemporaryTable("inputTable")

    val inputTable = blinkStreamTableEnv.from("inputTable")

    //转换操作
    //简单转换
    val resultTable = inputTable
      .select('id, 'temp)
      .filter('id === "sensor_1")


    val sql =
      """
        |select id, count(id) as cnt from inputTable group by id
        |""".stripMargin

    //聚合转换
    //    val aggTable = inputTable
    //      .groupBy('id)
    //      .select('id, 'id.count as 'cnt)
    val aggTable = blinkStreamTableEnv.sqlQuery(sql)

    val sinkDDL =
      """
        |create table jdbcOutputTable (
        | id string ,
        | cnt BIGINT not null
        |) with (
        | 'connector.type' = 'jdbc',
        | 'connector.url' = 'jdbc:mysql://localhost:3306/flink',
        | 'connector.table' = 'test',
        | 'connector.driver' = 'com.mysql.jdbc.Driver',
        | 'connector.username' = 'root',
        | 'connector.password' = '200028'
        |)
""".stripMargin

    blinkStreamTableEnv.sqlUpdate(sinkDDL)
    aggTable.insertInto("jdbcOutputTable")

    env.execute()

  }
}
